<?php
namespace console;

use database\MyKafka;

use config\KafkaConfig;

class KafkaController {


    public function consumer($paramArr = array()){

        $ptId = 0;
        if(count($paramArr) > 0){
            $ptId =  (int)$paramArr[0];
        }

        try {
            $consumer = MyKafka::getConsumerConn();
            $topicName =KafkaConfig::TEST_TOPIC;
            $consumer->setGroup(KafkaConfig::TEST_GROUP);
            $consumer->setPartition($topicName, $ptId);
            $consumer->setFromOffset(true);
            $consumer->setMaxBytes(102400);
        }catch (\Exception $e){
            echo $e->getMessage()."\r\n";exit;
        }

        while (true){
            try {

                $result = $consumer->fetch();

            } catch (\Kafka\Exception $e) {
                pre($e->getMessage()."\r\n");
            }

            foreach ($result as $topicName => $partition) {
                foreach ($partition as $partId => $messageSet) {
                    foreach ($messageSet as $message) {
                        $msgInfo = json_decode($message,true);
                        $msgType = empty($msgInfo["type"]) ? "" : $msgInfo["type"];
                        switch ($msgType){
                            case "test":
                                self::buss1($msgInfo);
                                break;
                        }
                    }
                }
            }


        }
    }

    public function test(){
        while (true){
            sendKafkaMsg();
            usleep(200);
        }
    }

    private function buss1($msgInfo){
        pre($msgInfo);
    }
}